package com.sub.kafka.demo;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName KafkaProducer
 * @Description: Kafka生产者案例
 * @Author Submerge.
 * @Since 2025/6/1 00:06
 * @Version 1.0
 */
public class KafkaProducerDemo {

    public static void main(String[] args) throws InterruptedException {

        //配置属性集合
        Map<String, Object> config =new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //创建KafkaProducer对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(config);

        //创建消息并发送
        for (int i = 0; i < 10; i++) {
            //创建消息对象，参数：Topic主题名称，Key，Value，当topic不存在时，会自动创建
            //也可以设置分区
            ProducerRecord<String, String> record = new ProducerRecord<>("sub-topic", "key"+i, "value"+i);
            Thread.sleep(1000);
            //生产者发送消息
            producer.send(record);
        }

        //关闭资源
        producer.close();

    }
}
